<?php

namespace Xin\WorkmanServer;

use Generator;
use Symfony\Component\HttpFoundation\Response as SymfonyResponse;
use Symfony\Component\HttpFoundation\StreamedResponse as SymfonyStreamResponse;
use Workerman\Connection\TcpConnection;
use Workerman\Protocols\Http\Request as WorkermanRequest;
use Workerman\Worker;
use Xin\Support\Reflect;

class HttpServer extends Server
{
	/**
	 * @var Worker
	 */
	protected $worker;

	/**
	 * @var callable
	 */
	protected $requestResolver;

	/**
	 * @var callable
	 */
	protected $httpHandler;

	/**
	 * @return callable
	 */
	public function getHttpHandler()
	{
		return $this->httpHandler;
	}

	/**
	 * @param callable $httpHandler
	 * @return void
	 */
	public function setHttpHandler(callable $httpHandler)
	{
		$this->httpHandler = $httpHandler;
	}


	/**
	 * @inerhitDoc
	 */
	public function run()
	{
		$this->config['handler'] = $this;
		$this->initServer();
	}

	/**
	 * @return Worker
	 */
	protected function initServer(): Worker
	{
		if (!$this->worker) {
			$bindAddress = $this->resolveBindAddress($this->config);
			$socketName = "http://" . $bindAddress;
			$context = $this->getConfig('context', []);
			$this->worker = tap(new Worker($socketName, $context), function ($worker) {
				WorkerUtil::bindConfig($worker, $this->config);
			});

			TcpConnection::$defaultMaxPackageSize = $this->getConfig('max_package_size') ?? 10 * 1024 * 1024;
		}

		return $this->worker;
	}

	/**
	 * @inerhitDoc
	 * @param TcpConnection $connection
	 * @param WorkermanRequest $workermanRequest
	 * @return void
	 */
	public function onMessage(TcpConnection $connection, WorkermanRequest $workermanRequest)
	{
		// 解析请求
		$request = $this->resolveRequest($workermanRequest);

		// 处理请求
		$isKeepAliveRequest = $this->isKeepAliveRequest($workermanRequest);
		$response = $this->handle($request, $isKeepAliveRequest);

		// 发送响应
		$this->sendResponse($connection, $request, $response, $isKeepAliveRequest);

		// 关闭连接
		if (
			(!$isKeepAliveRequest || $this->isAutoCloseConnectionFromRequest($request))
			&& !$this->isEventStreamResponse($response)
		) {
			$connection->close();
		}
	}

	/**
	 * @param mixed $request
	 * @param bool $isKeepAliveRequest
	 * @return mixed|SymfonyResponse
	 */
	protected function handle($request, bool $isKeepAliveRequest)
	{
		$response = call_user_func($this->httpHandler, $request, $isKeepAliveRequest);

		if ($response instanceof SymfonyResponse) {
			// 注入Server Name
			if (empty($response->headers->has('Server')) && $serverName = env('APP_NAME')) {
				$response->headers->set('Server', $serverName . " Server");
			}

			// 是保持活动请求
			if ($isKeepAliveRequest) {
				// 注入Connection
				$response->headers->set('Connection', 'keep-alive');

				// 判断ContentType类型
				if (!$this->isEventStreamResponse($response)) {
					// 注入Content-Length
					if (!$response->headers->has('Transfer-Encoding') && !$response->headers->has('Content-Length')) {
						$response->headers->set('Content-Length', strlen($response->getContent()));
					}
				}
			}
		}

		return $response;
	}

	/**
	 * 发送响应数据
	 * @param TcpConnection $connection
	 * @param mixed $request
	 * @param mixed $response
	 * @param bool $isKeepAliveRequest
	 * @return void
	 * @noinspection PhpDocMissingThrowsInspection
	 */
	protected function sendResponse(TcpConnection $connection, $request, $response, $isKeepAliveRequest)
	{
		$connection->send($response);
		if ($response instanceof SymfonyStreamResponse) {
			$callback = Reflect::get($response, 'callback');
			$data = $callback();
			if ($data instanceof Generator) {
				foreach ($data as $chunk) {
					if ($chunk instanceof SymfonyResponse) {
						$connection->send($chunk->getContent());
					} else {
						$connection->send($chunk);
					}
				}
			} else {
				$connection->send($callback());
			}
		}
	}

	/**
	 * 是否是事件流响应
	 * @param mixed $response
	 * @return bool
	 */
	protected function isEventStreamResponse($response)
	{
		if ($response instanceof SymfonyResponse) {
			return $response->headers->get('Content-Type') == 'text/event-stream';
		}

		return false;
	}

	/**
	 * @param WorkermanRequest $request
	 * @return mixed
	 */
	protected function resolveRequest(WorkermanRequest $request)
	{
		$server = $this->transformRequestHeaderToServer($request);

		return call_user_func($this->requestResolver, $request, $server);
	}

	/**
	 * 转换WorkermanRequest header到server
	 * @param WorkermanRequest $request
	 * @return array
	 */
	protected function transformRequestHeaderToServer(WorkermanRequest $request): array
	{
		$server = [];

		foreach ($request->header() as $key => $value) {
			$server['HTTP_' . strtoupper($key)] = $value;
		}

		return $server;
	}

	/**
	 * @param WorkermanRequest $workermanRequest
	 * @return bool
	 */
	public function isKeepAliveRequest(WorkermanRequest $workermanRequest): bool
	{
		$keepAlive = $workermanRequest->header('connection');

		return ($keepAlive === null && $workermanRequest->protocolVersion() === '1.1')
			|| $keepAlive === 'keep-alive' || $keepAlive === 'Keep-Alive';
	}

	/**
	 * 根据请求判断是否自动关闭连接
	 * @param mixed $request
	 * @return bool
	 */
	protected function isAutoCloseConnectionFromRequest($request): bool
	{
		if (method_exists($request, 'hasMacro') && $request::hasMacro('autoCloseConnection')) {
			return $request->getAutoCloseConnection();
		} elseif (method_exists($request, 'getAutoCloseConnection')) {
			return $request->getAutoCloseConnection();
		}

		return true;
	}

	/**
	 * @return callable|null
	 */
	public function getRequestResolver(): ?callable
	{
		return $this->requestResolver;
	}

	/**
	 * @param callable $requestResolver
	 */
	public function setRequestResolver(callable $requestResolver)
	{
		$this->requestResolver = $requestResolver;
	}

	/**
	 * @return Worker
	 */
	public function getWorker(): Worker
	{
		return $this->worker;
	}
}
